// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_SCAN_STRUCT_HPP_
#define _XBASE_SCAN_STRUCT_HPP_

#include <string>
#include <vector>
#include <map>
#include <set>

#include <boost/shared_ptr.hpp>

#include "error_code.hpp"
#include "binary_iarchive.hpp"

#include "protocol_struct.hpp"
#include "protocol_packet.hpp"

namespace xbase
{
	struct tablet_scan_info
	{
		tablet_scan_info(const std::string &tablet_name = "",
				 const std::string &start_row = "",
				 const std::string &end_row = "")
			: m_tablet_name(tablet_name), m_start_row(start_row), m_end_row(end_row),
			  m_latest_time_stamp(0), m_oldest_time_stamp(0), m_ttl(0), m_version_number(1),
			  m_options(0), m_num_row(1), m_row_offset(0), m_row_count(0)
		{}

		std::string m_tablet_name;
		std::string m_start_row;
		std::string m_end_row;
		std::vector<std::string> m_column_names;
		uint64_t m_latest_time_stamp;
		uint64_t m_oldest_time_stamp;
		uint32_t m_ttl;
		uint32_t m_version_number;
		uint32_t m_options;
		uint32_t m_num_row;
		uint32_t m_row_offset;
		uint32_t m_row_count;
	};

	struct scan_info
	{
		scan_info() {}

		scan_info(const std::string &table_name,
			  const std::string &start_row,
			  const std::string &end_row)
			: m_table_name(table_name), m_start_row(start_row), m_end_row(end_row),
			  m_latest_time_stamp(0), m_oldest_time_stamp(0), m_ttl(0), m_version_number(1),
			  m_options(0), m_num_row(1), m_row_offset(0), m_row_count(0)
		{}

		void add_column(const std::string &column_name)
		{
			m_column_names.push_back(column_name);
		}

		std::string m_table_name;
		std::string m_start_row;
		std::string m_end_row;
		std::vector<std::string> m_column_names;
		uint64_t m_latest_time_stamp;
		uint64_t m_oldest_time_stamp;
		uint32_t m_ttl;
		uint32_t m_version_number;
		uint32_t m_options;
		uint32_t m_num_row;
		uint32_t m_row_offset;
		uint32_t m_row_count;
	};

	struct cell
	{
		typedef xbase::protocol::key key;
		typedef std::string value;
		typedef xbase::protocol::record_vector record_vector;

		typedef std::pair<uint64_t, std::string> version;
		typedef std::vector<version> version_vector;

		cell(const std::string& column_name)
			: m_column_name(column_name)
		{}

		cell() {}

		void add_version(const version& v)
		{
			m_versions.push_back(v);
		}

		std::size_t get_version_count() const
		{
			return m_versions.size();
		}

		bool get_version(std::size_t i, version& v) const
		{
			if (i > m_versions.size())
			{
				return false;
			}
			
			v = m_versions[i];
			return true;
		}

		template <typename T>
		bool get_latest(T& t) const
		{
			typedef ::kb::network_byteorder byteorder;
			typedef ::kb::binary_iarchive<byteorder> iarchive;

			std::string latest = m_versions.front().second;
			iarchive ia(latest.c_str(), latest.length());

			return ia.load(&t);
		}

		version_vector m_versions;
		std::string m_column_name;
	};

	class scan_block
	{
	public:	
		scan_block() : m_eos_flag(false)
		{
			m_cur = m_records.begin();
		}

		void load(protocol::scan_next_response &resp)
		{
			if (resp.m_error_code == xbase::error::eos)
			{
				m_eos_flag = true;
			}

			//m_records = resp.m_records;
			m_records.swap(resp.m_records);
			m_cur = m_records.begin();
		}

		bool more() const
		{
			return (m_cur!=m_records.end());
		}

		protocol::record& front()
		{
			return *m_cur;
		}

		const protocol::record& front() const
		{
			return *m_cur;
		}

		void pop_front()
		{
			if (m_cur != m_records.end())
			{
				++m_cur;
			}
		}

		bool eof() const
		{
			return (!more()
				&& m_eos_flag);
		}
	private:
		bool m_eos_flag;
		protocol::record_vector m_records;
		protocol::record_vector::iterator m_cur;
	};

	struct row_result
	{
		typedef boost::shared_ptr<row_result> ptr_type;
		typedef std::vector<cell> cell_vector;

		row_result() {}

		void load(scan_block& sb)
		{
			m_cell_vector.clear();

			// 读取一行记录
			m_row_name = sb.front().m_key.m_row_name;

			while (sb.more()
				&& (m_row_name == sb.front().m_key.m_row_name))
			{
				// 读取一列
				cell c(sb.front().m_key.m_column_name);
				while (sb.more()
					&& (sb.front().m_key.m_row_name == m_row_name)
					&& (sb.front().m_key.m_column_name == c.m_column_name))
				{
					protocol::record &r = sb.front();
					cell::version v(r.m_key.m_timestamp, r.m_value);
					c.add_version(v);
					sb.pop_front();
				}
				m_cell_vector.push_back(c);
			}
		}

		bool get_cell(const std::string &column_name, cell& c) const
		{
			cell_vector::const_iterator iter = m_cell_vector.begin();
			while (iter != m_cell_vector.end())
			{
				if (iter->m_column_name == column_name)
				{
					c = *iter;
					return true;
				}
				++iter;
			}

			return false;
		}

		std::string m_row_name;
		cell_vector m_cell_vector;
	};

	typedef row_result::ptr_type row_result_ptr;

	struct meta_row_result
	{
		typedef boost::shared_ptr<meta_row_result> ptr_type;

		protocol::tablet_info m_tablet_info;
		protocol::server_address m_server_address;
		uint64_t m_server_start_code;
		std::string m_splitA;
		std::string m_splitB;

		bool load(row_result::ptr_type raw_ptr)
		{
			cell c;

			// tablet info
			std::string feild_name = "info:tabletInfo";
			if (raw_ptr->get_cell(feild_name, c))
			{
				if (!c.get_latest(m_tablet_info))
				{
					return false;
				}
			}

			// server address
			feild_name = "info:serverAddress";
			if (raw_ptr->get_cell(feild_name, c))
			{
				c.get_latest(m_server_address);
			}

			// server start code
			feild_name = "info:serverStartCode";
			if (raw_ptr->get_cell(feild_name, c))
			{
				c.get_latest(m_server_start_code);
			}

			// splitA
			feild_name = "info:splitA";
			if (raw_ptr->get_cell(feild_name, c))
			{
				c.get_latest(m_splitA);
			}

			// splitB
			feild_name = "info:splitB";
			if (raw_ptr	->get_cell(feild_name, c))
			{
				c.get_latest(m_splitB);
			}

			return true;
		}
	};

	typedef meta_row_result::ptr_type meta_row_result_ptr;
}

#endif
